package com.flume.demo1;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author shengtao
 * @Description:
 * @date 2019/01/11 18:18
 */
public class CustomInterceptor implements Interceptor {

    private final String headerKey;
    private static final String CONF_HEADER_KEY = "header";
    private static final String DEFAULT_HEADER = "count";
    private final AtomicLong currentCount;
    private Context ctx;

    public CustomInterceptor(Context ctx) {
        this.ctx = ctx;
        headerKey = ctx.getString(CONF_HEADER_KEY,DEFAULT_HEADER);
        currentCount = new AtomicLong();
    }

    //运行前的初始化，一般不需要实现
    @Override
    public void initialize() {

    }

    //)处理单个event
    @Override
    public Event intercept(Event event) {
        long count = currentCount.incrementAndGet();
        event.getHeaders().put(headerKey, String.valueOf(count));
        return event;
    }

    //)处理多个event
    @Override
    public List<Event> intercept(List<Event> list) {
        for(Event e:list) {
            intercept(e);
        }
        return list;
    }

    @Override
    public void close() {

    }
}
